package com.haozhen.streaming

/**
  * @author haozhen
  * @email haozh@ync1.com
  * @date 2021/1/16  21:09
  */
object BlackListFilter1 {

  def main(args: Array[String]): Unit = {
    import org.apache.log4j.{Level, Logger}
    import org.apache.spark.SparkConf
    import org.apache.spark.rdd.RDD
    import org.apache.spark.streaming.dstream.{ConstantInputDStream, DStream}
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    Logger.getLogger("org").setLevel(Level.WARN)

    val conf: SparkConf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val ssc = new StreamingContext(conf,Seconds(10))
    val blackList = Array(("spark",true),("scala",false),("hello",true))
        .filter(_._2).map(_._1.toLowerCase)
    val blackListRDD = ssc.sparkContext.makeRDD(blackList)

    val words = "hello world Hello jadoop jell spark scale scala java".split("\\s+").zipWithIndex
        .map{case(k,v)=>s"$v $k"}
      val rdd: RDD[String] = ssc.sparkContext.makeRDD(words)
    val wordDStream: ConstantInputDStream[String] = new ConstantInputDStream(ssc,rdd)

    //流式数据的处理
    //直接过滤掉数据
    wordDStream.map(line=>(line.split("\\s+")(1).toLowerCase,line))
        .filter{ case(word,timestamp) => !blackList.contains(word) }
        .map(_._2)
        .print()

//    wordDStream.map(line=>(line.split("\\s+")(1).toLowerCase,line))
//        .transform{
//          rdd=>
//            import org.apache.spark.sql.{DataFrame, SparkSession}
//            val session: SparkSession = SparkSession.builder().config(rdd.sparkContext.getConf).getOrCreate()
//            import session.implicits._
//            val frame1: DataFrame = rdd.toDF("word","line")
//            val frame2: DataFrame = blackListRDD.toDF("word","flag")
//            frame1.join(frame2,Seq("word"),"left_outer")
////              .filter($"flag").rdd
//              .filter("flag is null or flag = false").select("line").rdd
//        }.print()
//    wordDStream.transform{
//      rdd=>
//        val value = rdd.map{line=>(line.split("\\s+")(1).toLowerCase,line)}
//            .leftOuterJoin(blackListRDD)
//            .filter{case (_,(_,right))=> !right.getOrElse(false)}
//            .map{case (_,(left,_))=>left}
//        value
//    }.print()

    //输出

    //启动作业
    ssc.start()
    ssc.awaitTermination()
  }
}
